package avicit.bdp.dgs.metadata.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.dto.query.dds.ExecutorQuery;
import avicit.bdp.common.dto.request.dds.ProcessSchedulerDto;
import avicit.bdp.common.service.dto.CommonTypeDTO;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.CommonTypeService;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.database.adapter.BaseDBAdapter;
import avicit.bdp.common.utils.database.adapter.DbAdapterUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.ExecutorApi;
import avicit.bdp.dds.api.ProcessDefinitionApi;
import avicit.bdp.dds.api.SchedulerApi;
import avicit.bdp.dgs.metadata.dao.MetadataCollectTaskDao;
import avicit.bdp.dgs.metadata.dto.MetadataCollectTaskDTO;
import avicit.bdp.dgs.metadata.util.MetaDataProcessDefinitionUtil;
import avicit.platform6.api.system.ConvertColumnClient;
import avicit.platform6.api.system.SysUserClient;
import avicit.platform6.api.system.impl.SystemConstant;
import avicit.platform6.api.sysuser.dto.SysUser;
import avicit.platform6.commons.utils.BusinessUtil;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.context.ThreadContextHelper;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.rest.msg.QueryRespBean;
import com.alibaba.fastjson2.JSONArray;
import com.github.pagehelper.Page;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import javax.annotation.Resource;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * @author mayi 元数据采集任务服务层
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class MetadataCollectTaskService
    extends BaseService<MetadataCollectTaskDao, MetadataCollectTaskDTO> {

  private static final int ENABLE = 1;
  private static final int ZERO = 0;

  @Resource private MetadataCollectTaskDao metadataCollectTaskDao;

  @Resource private ProcessDefinitionApi processDefinitionApi;

  @Resource private ConvertColumnClient convertColumnClient;

  @Resource private SchedulerApi schedulerApi;

  @Resource private ExecutorApi executorApi;

  @Resource private DataSourceService dataSourceService;

  @Resource private CommonTypeService commonTypeService;

  @Autowired private SysUserClient sysUserClient;

  @Resource private MetadataTableColumnService metadataTableColumnService;
  @Resource private MetadataTableColumnPartitionService metadataTableColumnPartitionService;
  @Resource private MetadataTableHistoryService metadataTableHistoryService;
  @Resource private MetadataTableService metadataTableService;
  @Resource private MetadataTableColumnHistoryService metadataTableColumnHistoryService;

  @Resource
  private MetadataTableColumnPartitionHistoryService metadataTableColumnPartitionHistoryService;

  @Transactional(readOnly = true)
  public QueryRespBean<MetadataCollectTaskDTO> getPageList(
      String datasourceType, String keyWords, Integer pageNo, Integer pageSize) throws Exception {

    QueryRespBean<MetadataCollectTaskDTO> queryRespBean = new QueryRespBean<>();

    List<String> roleIdList =
        dataSourceService.getRoleListByUserId(ThreadContextHelper.getUserId());

    // 文档密级
    List<String> wordSecretList = ThreadContextHelper.getWordSecretList();
    com.github.pagehelper.PageHelper.startPage(pageNo, pageSize);
    Page<MetadataCollectTaskDTO> pageResult =
        metadataCollectTaskDao.getPageList(datasourceType, keyWords, roleIdList, wordSecretList);

    try {
      valueConvert(pageResult);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    queryRespBean.setResult(pageResult);
    BdpLogUtil.log4Query(queryRespBean);
    return queryRespBean;
  }

  private void valueConvert(Page<MetadataCollectTaskDTO> pageResult) {
    // 循环组装请求数据
    Map<String, Set<String>> convertFormData = new HashMap<>(16);
    for (MetadataCollectTaskDTO taskDTO : pageResult) {
      BusinessUtil.createConvertSet(convertFormData, SystemConstant.USER, taskDTO.getCreatedBy());
      BusinessUtil.createConvertSet(convertFormData, Constants.BDP_DATA_SECRET_LEVEL_KEY, taskDTO.getSecretLevel());
    }
    if (convertFormData.size() > 0) {
      // 获取请求结果
      Map<String, Map<String, String>> convertResultData =
          convertColumnClient.replace(convertFormData);
      // 循环设置Alias或Name的值
      for (MetadataCollectTaskDTO taskDTO : pageResult) {
        taskDTO.setCreatedBy(BusinessUtil.convertFormat(
                convertResultData, SystemConstant.USER, taskDTO.getCreatedBy()));
        taskDTO.setSecretLevelName(BusinessUtil.convertFormat(
                convertResultData, Constants.BDP_DATA_SECRET_LEVEL_KEY, taskDTO.getSecretLevel()));
      }
    }
  }

  public String save(MetadataCollectTaskDTO dto) {
    CheckParaUtils.checkStringValidity(dto.getName(), logger, "任务名称必填");
    CheckParaUtils.checkStringValidity(dto.getDatasourceType(), logger, "数据源类型必填");
    CheckParaUtils.checkStringValidity(dto.getDatasourceId(), logger, "数据源必填");

    dto.setId(ComUtil.getId());

    String datasourceId = dto.getDatasourceId();
    MetadataCollectTaskDTO query = new MetadataCollectTaskDTO();
    query.setDatasourceId(datasourceId);
    Long count = this.selectCount(query);
    if (count > 0) {
      throw new BusinessException("该数据源的元数据采集任务已存在");
    }

    checkDataSource(datasourceId);

    Map<String, Object> definitionMap =
        MetaDataProcessDefinitionUtil.generateProcessDefinition(dto);
    String processDefinitionId = processDefinitionApi.createProcessDefinition(definitionMap);
    if (StringUtils.isEmpty(processDefinitionId)) {
      throw new BusinessException("调度任务创建异常");
    }

    logger.info("create Process Definition, definitionId is  {}", processDefinitionId);
    dto.setProcessDefinitionId(processDefinitionId);

    this.insert(dto);
    log4Insert(dto);

    return dto.getId();
  }

  private void checkDataSource(String datasourceId) {
    DataSourceDTO bdpDasDatasource = dataSourceService.queryDataSourceById(datasourceId);
    if (bdpDasDatasource == null) {
      throw new BusinessException("该数据源的不存在");
    }

    // 检测数据源的连通性
    BaseDataSource baseDataSource = DBUtils.getBaseDataSource(bdpDasDatasource);
    try {
      BaseDBAdapter dbAdapter = DbAdapterUtils.getDbAdapter(baseDataSource.type());
      boolean flag = dbAdapter.dataSourceConnTest(baseDataSource);
      if (!flag) {
        throw new BusinessException("该数据源连接失败");
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  public void update(MetadataCollectTaskDTO dto) {

    CheckParaUtils.checkStringValidity(dto.getName(), logger, "任务名称必填");
    CheckParaUtils.checkStringValidity(dto.getDatasourceType(), logger, "数据源类型必填");
    CheckParaUtils.checkStringValidity(dto.getDatasourceId(), logger, "数据源必填");

    MetadataCollectTaskDTO metadataCollectTaskDTO = this.selectByPrimaryKey(dto.getId());
    CheckParaUtils.checkDbObjectValidity(
        metadataCollectTaskDTO, logger, String.format("没有查到数据，请确保id:%s的数据是否存在", dto.getId()));
    dto.setCreationDate(metadataCollectTaskDTO.getCreationDate());

    checkDataSource(dto.getDatasourceId());

    Map<String, Object> definitionMap =
        MetaDataProcessDefinitionUtil.generateProcessDefinition(dto);
    definitionMap.put("id", dto.getProcessDefinitionId());
    Boolean result = processDefinitionApi.updateProcessDefinition(definitionMap);
    logger.info("update Process Definition result {}", result);

    this.updateByPrimaryKey(dto);
    BdpLogUtil.log4Update(dto, metadataCollectTaskDTO);
  }

  public Boolean batchDelete(String ids) {
    CheckParaUtils.checkStringValidity(ids, logger, "没有传入待删除的ID");
    String[] idArr = ids.split(Constants.COMMA);
    for (String id : idArr) {
      MetadataCollectTaskDTO dto = this.selectByPrimaryKey(id);
      CheckParaUtils.checkDbObjectValidity(
          dto, logger, String.format("没有查到数据，请确保id:%s的数据是否存在", id));

      try {
        Boolean result = processDefinitionApi.deleteProcessDefinition(dto.getProcessDefinitionId());
        logger.info("delete Process Definition result {}", result);
      } catch (Exception e) {
        logger.error("delete Process Definition failed", e);
      }

      String datasourceId = dto.getDatasourceId();

      // 删除采集任务数据源下所有的元数据
      metadataTableColumnService.deleteByDatasourceId(datasourceId);
      metadataTableColumnPartitionService.deleteByDatasourceId(datasourceId);
      metadataTableColumnHistoryService.deleteByDatasourceId(datasourceId);
      metadataTableColumnPartitionHistoryService.deleteByDatasourceId(datasourceId);
      metadataTableHistoryService.deleteByDatasourceId(datasourceId);
      metadataTableService.deleteByDatasourceId(datasourceId);

      this.mapper.deleteByPrimaryKey(id);
      // 记录日志
      BdpLogUtil.log4Delete(dto);
    }
    return true;
  }

  public MetadataCollectTaskDTO queryById(String id) {
    MetadataCollectTaskDTO dto = this.selectByPrimaryKey(id);
    CheckParaUtils.checkDbObjectValidity(dto, logger, String.format("没有查到数据，请确保id:%s的数据是否存在", id));
    if (StringUtils.isNotEmpty(dto.getDatasourceId())) {
      DataSourceDTO sourceDTO = dataSourceService.selectByPrimaryKey(dto.getDatasourceId());
      if (sourceDTO != null) {
        dto.setDatasourceName(sourceDTO.getDatasourceName());
      }
    }
    if (StringUtils.isNotBlank(dto.getTableBelongCatalogId())) {
      CommonTypeDTO commonTypeDTO =
          commonTypeService.selectByPrimaryKey(dto.getTableBelongCatalogId());
      if (commonTypeDTO != null) {
        dto.setTableBelongCatalogName(commonTypeDTO.getName());
      }
    }

    if (StringUtils.isNotBlank(dto.getOwner())) {
      try {
        JSONArray array = JSONUtils.parseArray(dto.getOwner());
        List<String> idList = array.toJavaList(String.class);
        List<SysUser> userList = sysUserClient.getUserListByIdList(idList);
        if (CollectionUtils.isNotEmpty(userList)) {
          List<String> userNameList =
              userList.stream().map(SysUser::getName).collect(Collectors.toList());
          dto.setOwnerName(String.join(",", userNameList));
        }
      } catch (Exception e) {
        logger.error("解析owner失败", e);
      }
    }

    return dto;
  }

  /**
   * 启动任务
   *
   * @param processDefinitionId 流程ID
   * @return 执行状态
   */
  public Boolean startTaskByProcessDefinitionId(String processDefinitionId) {
    ExecutorQuery executorQuery = new ExecutorQuery();
    executorQuery.setProcessDefinitionId(processDefinitionId);
    executorQuery.setWarningType(ZERO);
    executorQuery.setFailureStrategy(ZERO);
    executorQuery.setExecType(ZERO);
    executorQuery.setRunMode(ENABLE);

    boolean result = executorApi.startProcessInstance(executorQuery);
    if (result) {
      logger.info("task start success, processDefinitionId:{}", processDefinitionId);
    } else {
      logger.info("task start failed, processDefinitionId:{}", processDefinitionId);
      throw new BusinessException(String.format("task[%s] start  failed", processDefinitionId));
    }

    return true;
  }

  /**
   * 停止、暂停作业
   *
   * @param processInstanceId 流程id
   * @param executeType 执行类型
   * @return 执行结果
   */
  public Boolean executeByProcessInstanceId(String processInstanceId, String executeType) {
    boolean result = executorApi.executeByProcessInstanceId(processInstanceId, executeType);
    if (result) {
      logger.info("task execute success, processInstanceId:{}", processInstanceId);
    } else {
      logger.info("task execute failed, processInstanceId:{}", processInstanceId);
      throw new BusinessException(
          String.format(
              "task execute failed,executeType is " + "%s, taskId is %s",
              executeType, processInstanceId));
    }
    return true;
  }

  /**
   * 根据taskId查看定时任务用来页面回显
   *
   * @param processDefinitionId 流程ID
   */
  public Map<String, Object> getScheduleByProcessDefinitionIdId(String processDefinitionId) {
    List<Map<String, Object>> list = this.getScheduleListByProcessDefinitionId(processDefinitionId);
    if (CollectionUtils.isEmpty(list)) {
      return null;
    }

    Map<String, Object> map = list.get(0);
    map.put("warningType", map.get("warningType"));
    map.put("failureStrategy", map.get("failureStrategy"));
    map.put("processInstancePriority", map.get("processInstancePriority"));
    return map;
  }

  /**
   * 根据流程ID获取定时任务列表
   *
   * @param processDefinitionId 流程ID
   * @return 定时任务列表
   */
  public List<Map<String, Object>> getScheduleListByProcessDefinitionId(
      String processDefinitionId) {
    QueryRespBean<Map<String, Object>> mapQueryRespBean =
        schedulerApi.queryScheduleListPaging(processDefinitionId, null, 1, 10);
    return mapQueryRespBean.getResult().getResult();
  }

  /**
   * 保存&修改定时任务
   *
   * @param processSchedulerDto 定时任务参数
   */
  public Boolean saveOrUpdateSchedule(ProcessSchedulerDto processSchedulerDto) {
    CheckParaUtils.checkStringValidity(
        processSchedulerDto.getProcessDefinitionId(), logger, "流程定义ID不能为空！");
    CheckParaUtils.checkObjectValidity(processSchedulerDto.getStartTime(), logger, "开始时间不能为空！");
    CheckParaUtils.checkObjectValidity(processSchedulerDto.getEndTime(), logger, "结束时间不能为空！");
    CheckParaUtils.checkStringValidity(
        processSchedulerDto.getCrontab(), logger, "crontab时间表达式不能为空！");

    // 判断这个流程下的定时任务个数，如果有则修改，没有则添加
    List<Map<String, Object>> result =
        this.getScheduleListByProcessDefinitionId(processSchedulerDto.getProcessDefinitionId());
    boolean success;
    if (CollectionUtils.isEmpty(result)) {
      success = schedulerApi.createSchedule(processSchedulerDto);
      if (success) {
        logger.info("新增定时任务成功, {}", processSchedulerDto);
      } else {
        logger.info("新增定时任务失败, {}", processSchedulerDto);
      }
    } else {
      CheckParaUtils.checkStringValidity(processSchedulerDto.getId(), logger, "定时设置ID不能为空");
      success = schedulerApi.updateSchedule(processSchedulerDto);
      if (success) {
        logger.info("修改定时任务成功, {}", processSchedulerDto);
      } else {
        logger.info("修改定时任务失败, {}", processSchedulerDto);
      }
    }

    if (success) {
      int cronTaskNum = 0;
      List<Map<String, Object>> cronTaskList =
          this.getScheduleListByProcessDefinitionId(processSchedulerDto.getProcessDefinitionId());
      if (CollectionUtils.isNotEmpty(cronTaskList)) {
        cronTaskNum = cronTaskList.size();
      }
      this.metadataCollectTaskDao.updateCrontabTaskNum(
          processSchedulerDto.getProcessDefinitionId(), cronTaskNum);
    }

    return success;
  }

  /**
   * 启用/停用任务
   *
   * @param id 任务ID
   * @param enable 启用停用状态
   */
  public boolean releaseTask(String id, Integer enable) {
    MetadataCollectTaskDTO taskDTO = this.selectByPrimaryKey(id);
    CheckParaUtils.checkDbObjectValidity(
        taskDTO, logger, String.format("没有查到数据，请确保id:%s的任务数据是否存在", id));
    taskDTO.setEnable(enable);
    // 启用定时任务，是：1，否：0
    taskDTO.setEnableCronTab(enable);

    processDefinitionApi.releaseProcessDefinition(taskDTO.getProcessDefinitionId(), enable);
    this.updateByPrimaryKeySelective(taskDTO);
    return true;
  }
}
